package com.raptor.logs;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.classic.spi.IThrowableProxy;
import ch.qos.logback.classic.spi.StackTraceElementProxy;
import ch.qos.logback.core.Appender;
import ch.qos.logback.core.spi.AppenderAttachableImpl;
import com.shihang.core.base.util.SerializationUtil;
import com.raptor.logs.delivery.FailedDeliveryCallback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaAppender extends KafkaAppenderConfig<ILoggingEvent> {

    private LazyProducer lazyProducer = null;
    private final AppenderAttachableImpl<ILoggingEvent> aai = new AppenderAttachableImpl<ILoggingEvent>();
    private final FailedDeliveryCallback<ILoggingEvent> failedDeliveryCallback = new FailedDeliveryCallback<ILoggingEvent>() {
        @Override
        public void onFailedDelivery(ILoggingEvent evt, Throwable throwable) {
            aai.appendLoopOnAppenders(evt);
        }
    };

    SimpleDateFormat sdff = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
    ExecutorService executorService = Executors.newFixedThreadPool(2);

    public KafkaAppender() {
        addProducerConfigValue("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        addProducerConfigValue("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }

    @Override
    public void doAppend(ILoggingEvent e) {
//        ensureDeferredAppends();
//        if (e instanceof ILoggingEvent) {
//            deferAppend(e);
//        } else {
        super.doAppend(e);
//        }
    }

    @Override
    public void start() {
        // only error free appenders should be activated
        if (!checkPrerequisites()) return;

//        if (partition != null && partition < 0) {
//            partition = null;
//        }

        lazyProducer = new LazyProducer();

        super.start();
    }

    @Override
    public void stop() {
        super.stop();
        if (lazyProducer != null && lazyProducer.isInitialized()) {
            try {
                lazyProducer.get().close();
            } catch (KafkaException e) {
                this.addWarn("Failed to shut down kafka producer: " + e.getMessage(), e);
            }
            lazyProducer = null;
        }
    }

//    @Override
//    public void setKeyingStrategy(KeyingStrategy<? super ILoggingEvent> keyingStrategy) {
//        super.setKeyingStrategy(new LoggerNameKeyingStrategy());
//    }


    @Override
    public void addAppender(Appender<ILoggingEvent> newAppender) {
        aai.addAppender(newAppender);
    }

    @Override
    public Iterator<Appender<ILoggingEvent>> iteratorForAppenders() {
        return aai.iteratorForAppenders();
    }

    @Override
    public Appender<ILoggingEvent> getAppender(String name) {
        return aai.getAppender(name);
    }

    @Override
    public boolean isAttached(Appender<ILoggingEvent> appender) {
        return aai.isAttached(appender);
    }

    @Override
    public void detachAndStopAllAppenders() {
        aai.detachAndStopAllAppenders();
    }

    @Override
    public boolean detachAppender(Appender<ILoggingEvent> appender) {
        return aai.detachAppender(appender);
    }

    @Override
    public boolean detachAppender(String name) {
        return aai.detachAppender(name);
    }


    void buildLogInfo(Object[] parameters, LogInfo info) {
        if (parameters == null || info == null)
            return;
        if (parameters.length > 0) {
            if (parameters[0] instanceof Integer) {
                info.setMsec((Integer) parameters[0]);
            }
        }
    }

    String getStackTraceInfo(StackTraceElementProxy[] stackTraceElementProxies) {
        if (stackTraceElementProxies == null)
            return "";

        StringBuilder sb = new StringBuilder();
        for (StackTraceElementProxy stackTraceElementProxy : stackTraceElementProxies) {
            StackTraceElement stackTraceElement = stackTraceElementProxy.getStackTraceElement();
            sb.append(stackTraceElement.toString()).append("\n");
        }

        return sb.toString();
    }

    @Override
    protected void append(ILoggingEvent e) {
        String topic = "";
        LogInfo logInfo = new LogInfo();
        String message = e.getMessage();
        IThrowableProxy proxy = e.getThrowableProxy();
        Object[] argArray = e.getArgumentArray();
        switch (e.getLevel().levelStr) {
            case "ERROR":
                topic = "monitor-error-log";
                logInfo.setMessage(message);
                if (proxy != null) {
                    logInfo.setException(getStackTraceInfo(proxy.getStackTraceElementProxyArray()));
                }
                break;
            case "INFO":
                topic = "monitor-info-log";
                logInfo.setMessage(message);
                buildLogInfo(argArray, logInfo);
                break;
            case "WARN":
                topic = "monitor-warn-log";
                logInfo.setMessage(message);
                if (argArray != null && argArray.length > 0) {
                    Object arg =  argArray[0];
                    if (arg instanceof Integer) {
                        logInfo.setMsec((Integer) arg);
                    }
                }
                break;
            case "DEBUG":
                topic = "monitor-debug-log";
                logInfo.setMessage(message);
                buildLogInfo(argArray, logInfo);
                break;
            case "TRACE":
                topic = "monitor-trace-log";
                logInfo.setMessage(message);
                buildLogInfo(argArray, logInfo);
                break;
            default:
                topic = "monitor-other-log";
                break;
        }

        //build appName
        String appName = System.getenv("APP_NAME");
        if (appName.equals("")) {
            appName = getLogType();
        }

        //build ip
        String podIp = System.getenv("POD_IP");
        String hostIp = System.getenv("APP_HOST_IP");
        String ip = "";
        if (podIp != null && hostIp != null) {
            ip = hostIp + "-" + podIp;
        }

        logInfo.setLogType(appName);
        logInfo.setLevel(e.getLevel().levelStr);

        logInfo.setCreateTime(sdff.format(new Date(getTimestamp(e))) + "+08:00");
        logInfo.setLogger(e.getLoggerName());
        logInfo.setUserIp(ip);

        final Producer<String, String> producer = lazyProducer.get();
        String finalTopic = topic;
        executorService.submit(() -> {
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(finalTopic,
                    logInfo.getLogger(), SerializationUtil.serialize(logInfo));

            if (producer != null) {
                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        failedDeliveryCallback.onFailedDelivery(e, exception);
                    }
                });
            } else {
                failedDeliveryCallback.onFailedDelivery(e, null);
            }
        });

//        final ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(topic,key, payload);
//
//        final Producer<byte[], byte[]> producer = lazyProducer.get();
//        if (producer != null) {
//            deliveryStrategy.send(lazyProducer.get(), record, e, failedDeliveryCallback);
//        } else {
//            failedDeliveryCallback.onFailedDelivery(e, null);
//        }
    }

    protected Long getTimestamp(ILoggingEvent e) {
        if (e != null) {
            return e.getTimeStamp();
        } else {
            return System.currentTimeMillis();
        }
    }

    protected Producer<String, String> createProducer() {
        return new KafkaProducer<>(this.producerConfig);
    }

//    private void deferAppend(ILoggingEvent event) {
//        queue.add(event);
//    }
//
//    // drains queue events to super
//    private void ensureDeferredAppends() {
//        ILoggingEvent event;
//
//        while ((event = queue.poll()) != null) {
//            super.doAppend(event);
//        }
//    }

    private class LazyProducer {

        private volatile Producer<String, String> producer;

        public Producer<String, String> get() {
            Producer<String, String> result = this.producer;
            if (result == null) {
                synchronized (this) {
                    result = this.producer;
                    if (result == null) {
                        this.producer = result = this.initialize();
                    }
                }
            }

            return result;
        }

        protected Producer<String, String> initialize() {
            Producer<String, String> producer = null;
            try {
                producer = createProducer();
            } catch (Exception e) {
                addError("error creating producer", e);
            }
            return producer;
        }

        public boolean isInitialized() {
            return producer != null;
        }
    }
}
